package com.digiwin.demo.mq;

import com.rabbitmq.client.BuiltinExchangeType;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.apache.flink.api.common.serialization.DeserializationSchema;

import java.io.IOException;
import java.io.Serializable;

/**
 * @Author yanggld
 * @Date 2019/12/23-14:50
 * @Description
 */
public class MyRMQSource<OUT>  extends RMQSource implements Serializable {

	public MyRMQSource(RMQConnectionConfig rmqConnectionConfig, String queueName, DeserializationSchema deserializationSchema) {
		super(rmqConnectionConfig, queueName, deserializationSchema);
	}

	public MyRMQSource(RMQConnectionConfig rmqConnectionConfig, String queueName, boolean usesCorrelationId, DeserializationSchema deserializationSchema) {
		super(rmqConnectionConfig, queueName, usesCorrelationId, deserializationSchema);
	}

	private String exchangeName;
	private BuiltinExchangeType exchangeType;
	private String routingKey;

	@Override
	protected void setupQueue() throws IOException {
		// 表示声明了一个队列
		channel.queueDeclare(queueName, true, false, false, null);
		// 表示声明了一个交换机
		channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
		// 建立一个绑定关系:
		channel.queueBind(queueName, exchangeName, routingKey);
	}

	public void setExchangeName(String exchangeName) {
		this.exchangeName = exchangeName;
	}

	public void setExchangeType(BuiltinExchangeType exchangeType) {
		this.exchangeType = exchangeType;
	}

	public void setRoutingKey(String routingKey) {
		this.routingKey = routingKey;
	}
}
